package org.example.infrastructure.redis;


import cn.hutool.json.JSONUtil;
import org.example.types.enums.ResponseCode;
import org.example.types.exception.AppException;
import org.redisson.api.*;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static org.example.types.common.RedisConstants.LOCK_PREFIX;

/**
 * @author yinghuashen
 * @version 1.0
 * @description
 * @date 2025/1/25 13:14
 */

@Service
public class RedisService implements IRedisService, DisposableBean {

    @Resource
    private RedissonClient redissonClient;

    private final static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,
            10, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),
            new ThreadPoolExecutor.CallerRunsPolicy());

    public <T> void setValue(String key, T value) {
        redissonClient.<T>getBucket(key).set(value);
    }

    @Override
    public <T> void setValue(String key, T value, long expired) {
        RBucket<T> bucket = redissonClient.getBucket(key);
        bucket.set(value, expired, TimeUnit.MINUTES);
    }

    public <T> T getValue(String key) {
        return redissonClient.<T>getBucket(key).get();
    }

    @Override
    public <T> RQueue<T> getQueue(String key) {
        return redissonClient.getQueue(key);
    }

    @Override
    public <T> RBlockingQueue<T> getBlockingQueue(String key) {
        return redissonClient.getBlockingQueue(key);
    }

    @Override
    public <T> RDelayedQueue<T> getDelayedQueue(RBlockingQueue<T> rBlockingQueue) {
        return redissonClient.getDelayedQueue(rBlockingQueue);
    }

    @Override
    public void setAtomicLong(String key, long value) {
        redissonClient.getAtomicLong(key).set(value);
    }

    @Override
    public Long getAtomicLong(String key) {
        return redissonClient.getAtomicLong(key).get();
    }

    @Override
    public long incr(String key) {
        return redissonClient.getAtomicLong(key).incrementAndGet();
    }

    @Override
    public long incrBy(String key, long delta) {
        return redissonClient.getAtomicLong(key).addAndGet(delta);
    }

    @Override
    public long decr(String key) {
        return redissonClient.getAtomicLong(key).decrementAndGet();
    }

    @Override
    public long decrBy(String key, long delta) {
        return redissonClient.getAtomicLong(key).addAndGet(-delta);
    }

    @Override
    public void remove(String key) {
        redissonClient.getBucket(key).delete();
    }

    @Override
    public boolean isExists(String key) {
        return redissonClient.getBucket(key).isExists();
    }

    public void addToSet(String key, String value) {
        RSet<String> set = redissonClient.getSet(key);
        set.add(value);
    }

    public boolean isSetMember(String key, String value) {
        RSet<String> set = redissonClient.getSet(key);
        return set.contains(value);
    }

    public void addToList(String key, String value) {
        RList<String> list = redissonClient.getList(key);
        list.add(value);
    }

    public String getFromList(String key, int index) {
        RList<String> list = redissonClient.getList(key);
        return list.get(index);
    }

    @Override
    public <K, V> RMap<K, V> getMap(String key) {
        return redissonClient.getMap(key);
    }

    public void addToMap(String key, String field, String value) {
        RMap<String, String> map = redissonClient.getMap(key);
        map.put(field, value);
    }

    public String getFromMap(String key, String field) {
        RMap<String, String> map = redissonClient.getMap(key);
        return map.get(field);
    }

    @Override
    public <K, V> V getFromMap(String key, K field) {
        return redissonClient.<K, V>getMap(key).get(field);
    }

    public void addToSortedSet(String key, String value) {
        RSortedSet<String> sortedSet = redissonClient.getSortedSet(key);
        sortedSet.add(value);
    }

    @Override
    public RLock getLock(String key) {
        return redissonClient.getLock(key);
    }

    @Override
    public RLock getFairLock(String key) {
        return redissonClient.getFairLock(key);
    }

    @Override
    public RReadWriteLock getReadWriteLock(String key) {
        return redissonClient.getReadWriteLock(key);
    }

    @Override
    public RSemaphore getSemaphore(String key) {
        return redissonClient.getSemaphore(key);
    }

    @Override
    public RPermitExpirableSemaphore getPermitExpirableSemaphore(String key) {
        return redissonClient.getPermitExpirableSemaphore(key);
    }

    @Override
    public RCountDownLatch getCountDownLatch(String key) {
        return redissonClient.getCountDownLatch(key);
    }

    @Override
    public <T> RBloomFilter<T> getBloomFilter(String key) {
        return redissonClient.getBloomFilter(key);
    }

    @Override
    public Boolean expire(String key, long expired, TimeUnit timeUnit) {
        try {
            // 获取 RBucket 对象，RBucket 用于操作单个值
            RBucket<Object> bucket = redissonClient.getBucket(key);
            // 尝试设置新的过期时间
            return bucket.expire(expired, timeUnit);
        } catch (Exception e) {
            // 发生异常时打印错误信息
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public Boolean setNx(String key) {
        return redissonClient.getBucket(key).trySet("lock");
    }

    @Override
    public Boolean setNx(String key, long expired, TimeUnit timeUnit) {
        return redissonClient.getBucket(key).trySet("lock", expired, timeUnit);
    }

    @Override
    public RBitSet getBitSet(String key) {
        return redissonClient.getBitSet(key);
    }

    @Override
    public void setCacheObjectWithLogicalExpire(String key, Object value, Long time, TimeUnit timeUnit) {
        // 设置逻辑过期
        RedisData redisData = new RedisData();
        redisData.setData(value);
        // 将 LocalDateTime 转换为 Date
        LocalDateTime expireDateTime = LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time));
        Date expireDate = Date.from(expireDateTime.atZone(ZoneId.systemDefault()).toInstant());
        redisData.setExpireTime(expireDate);        // 写入 Redis
        setValue(key, redisData);
    }

    @Override
    // TODO 加锁的时候应该设置过期时间。解锁时watchDog，查看是否是自己的锁。
    public <R, T> R getCacheObjectWithLogicalExpire(String key, T req, Class<R> type, Function<T, R> function, Long time, TimeUnit timeUnit) {
        // 1. 从 Redis 中获取缓存数据
        RedisData redisData = getValue(key);
        // 2. 判断缓存数据是否存在
        if (redisData != null) {
//            RedisData redisData = JSONUtil.toBean(json, RedisData.class);
            // 3. 判断是否过期
            Date expireTime = redisData.getExpireTime();
            // 4. 未过期，进行后续处理
            if (expireTime.after(new Date())) {
                String dataStr = (String) redisData.getData();
                if ("null".equals(dataStr)) {
                    return null;
                }
                return JSONUtil.toBean(dataStr, type);
            }
            // 5. 已过期，需要查询数据库
        }
        // 4. 不存在，或者已过期，需要查询数据库
        // TODO 完善点：要加自动过期锁，值设置为线程id
        RLock rLock = getLock(LOCK_PREFIX + key);
        if(rLock.tryLock()){
            try {
                // TODO 拓展点：再次判断缓存是否过期。因为极端情况下，有可能锁1释放的时候，锁2正好重新加锁。此时缓存已经重建好。
                threadPoolExecutor.submit(()->{
                    Thread.currentThread().setName("RebuildCache-Thread");
                    try {
                        // 4.1. 查询数据库
                        R newRecord = function.apply(req);
                        if (newRecord == null) {
                            // 缓存空值，设置较短过期时间
                            setCacheObjectWithLogicalExpire(key, "null", 1L, TimeUnit.MINUTES);
                        } else {
                            // 4.2. 写入 Redis
                            String jsonStr = JSONUtil.toJsonStr(newRecord);
                            setCacheObjectWithLogicalExpire(key, jsonStr, time, timeUnit);
                        }
                    }catch (Exception e){
                        throw new AppException(ResponseCode.CACHE_BUILD_ERROR);
                    }
                });
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                if (rLock.isHeldByCurrentThread()) {
                    rLock.unlock();
                }
            }
        }
        // 5. 已逻辑过期/未存在缓存中，数据无效，返回null
        return null;
    }

    @Override
    public void destroy() throws Exception {
        // 关闭线程池
        threadPoolExecutor.shutdown();
        try {
            // 等待线程池中的任务完成，最多等待 10 秒
            if (!threadPoolExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                // 如果 10 秒后仍有任务未完成，则强制关闭线程池
                threadPoolExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            // 如果等待过程中被中断，则强制关闭线程池
            threadPoolExecutor.shutdownNow();
        }
    }
}

